package cn.dwj.hadoop.mr.flowSumSort;

import cn.dwj.hadoop.mr.flowsum.FlowBean;
import cn.dwj.hadoop.mr.flowsum.FlowSumMapper;
import cn.dwj.hadoop.mr.flowsum.FlowSunReduce;
import cn.dwj.hadoop.mr.flowsum.FlowSunRunner;
import org.apache.commons.io.output.NullWriter;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * Created by Administrator on 2017/12/27 0027.
 */
public class SortMR {
    public static class SortMapper extends Mapper<LongWritable, Text, FlowBean, NullWritable>{

        //拿到一行数据，切分出各字段，封装为一个flowbean，作为key输出
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = StringUtils.split(line, "\t");
            String phoneNB = fields[0];
            long uFlow = Long.parseLong(fields[1]);
            long dFlow = Long.parseLong(fields[2]);

            context.write(new FlowBean(phoneNB, uFlow,dFlow), NullWritable.get());
        }

        public static class SortReduce extends Reducer<FlowBean, NullWritable, FlowBean, NullWritable>{
            @Override
            protected void reduce(FlowBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
                context.write(key, NullWritable.get());
            }
        }

        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);
            //设置整个job所在的那些类在哪个jar包中
            job.setJarByClass(SortMR.class);
            //本job使用的map和reduce的类
            job.setMapperClass(SortMapper.class);
            job.setReducerClass(SortReduce.class);

            //指定输出数据的类型，对mapper和reduce的类型都起作用
            job.setOutputKeyClass(FlowBean.class);
            job.setOutputValueClass(NullWritable.class);

            FileInputFormat.setInputPaths(job, new Path[]{new Path(args[0])});
            FileOutputFormat.setOutputPath(job, new Path(args[1]));//指定处理结果输出数据存放在路径

            System.exit(job.waitForCompletion(true)?0:1);
        }
    }
}
